// PricingStrategyReducer.java
package com.taxi.analysis.pricing;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 智能定价策略分析Reducer
 * 功能：使用高级聚合技术进行定价策略深度分析
 * 技术：高级聚合技术（多维度聚合、嵌套聚合、条件聚合、滑动窗口聚合、分层聚合）
 */
public class PricingStrategyReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        String keyStr = key.toString();

        if (keyStr.startsWith("MULTI_")) {
            processMultiDimensionalAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("NESTED_")) {
            processNestedAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("CONDITIONAL_")) {
            processConditionalAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("SLIDING_")) {
            processSlidingWindowAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("HIERARCHICAL_")) {
            processHierarchicalAggregation(keyStr, values, context);
        }
    }

    /**
     * 处理多维度聚合
     */
    private void processMultiDimensionalAggregation(String aggregateKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<MultiDimensionalData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("MULTI_DIMENSIONAL|")) {
                    String data = record.substring("MULTI_DIMENSIONAL|".length());
                    MultiDimensionalData mdData = parseMultiDimensionalData(data);
                    if (mdData != null) {
                        dataList.add(mdData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析多维度数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            // 多维度聚合分析
            MultiDimensionalAnalysis analysis = analyzeMultiDimensionalData(dataList);

            // 输出多维度分析结果
            outputMultiDimensionalAnalysis(aggregateKey, analysis, context);
        }
    }

    /**
     * 处理嵌套聚合
     */
    private void processNestedAggregation(String nestedKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<NestedData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("NESTED_L1|") || record.startsWith("NESTED_L2|") || record.startsWith("NESTED_L3|")) {
                    String[] parts = record.split("\\|", 2);
                    String level = parts[0];
                    String data = parts[1];
                    NestedData nestedData = parseNestedData(level, data);
                    if (nestedData != null) {
                        dataList.add(nestedData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析嵌套数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            // 嵌套聚合分析
            NestedAnalysis analysis = analyzeNestedData(dataList);

            // 输出嵌套分析结果
            outputNestedAnalysis(nestedKey, analysis, context);
        }
    }

    /**
     * 处理条件聚合
     */
    private void processConditionalAggregation(String conditionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<ConditionalData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("CONDITIONAL|")) {
                    String data = record.substring("CONDITIONAL|".length());
                    ConditionalData condData = parseConditionalData(data);
                    if (condData != null) {
                        dataList.add(condData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析条件数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            // 条件聚合分析
            ConditionalAnalysis analysis = analyzeConditionalData(dataList);

            // 输出条件分析结果
            outputConditionalAnalysis(conditionKey, analysis, context);
        }
    }

    /**
     * 处理滑动窗口聚合
     */
    private void processSlidingWindowAggregation(String windowKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<SlidingWindowData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("SLIDING_WINDOW|")) {
                    String data = record.substring("SLIDING_WINDOW|".length());
                    SlidingWindowData swData = parseSlidingWindowData(data);
                    if (swData != null) {
                        dataList.add(swData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析滑动窗口数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            // 滑动窗口聚合分析
            SlidingWindowAnalysis analysis = analyzeSlidingWindowData(dataList);

            // 输出滑动窗口分析结果
            outputSlidingWindowAnalysis(windowKey, analysis, context);
        }
    }

    /**
     * 处理分层聚合
     */
    private void processHierarchicalAggregation(String hierarchyKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<HierarchicalData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("HIERARCHICAL|")) {
                    String data = record.substring("HIERARCHICAL|".length());
                    HierarchicalData hData = parseHierarchicalData(data);
                    if (hData != null) {
                        dataList.add(hData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析分层数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            // 分层聚合分析
            HierarchicalAnalysis analysis = analyzeHierarchicalData(dataList);

            // 输出分层分析结果
            outputHierarchicalAnalysis(hierarchyKey, analysis, context);
        }
    }

    /**
     * 分析多维度数据
     */
    private MultiDimensionalAnalysis analyzeMultiDimensionalData(List<MultiDimensionalData> dataList) {
        MultiDimensionalAnalysis analysis = new MultiDimensionalAnalysis();

        double totalFare = 0, totalTip = 0, totalAmount = 0, totalDistance = 0, totalDuration = 0;
        double totalFarePerMile = 0, totalTipRate = 0, totalValueScore = 0;

        for (MultiDimensionalData data : dataList) {
            totalFare += data.fareAmount;
            totalTip += data.tipAmount;
            totalAmount += data.totalAmount;
            totalDistance += data.tripDistance;
            totalDuration += data.tripDuration;
            totalFarePerMile += data.farePerMile;
            totalTipRate += data.tipRate;
            totalValueScore += data.valueScore;
        }

        analysis.recordCount = dataList.size();
        analysis.avgFare = totalFare / dataList.size();
        analysis.avgTip = totalTip / dataList.size();
        analysis.avgTotalAmount = totalAmount / dataList.size();
        analysis.avgDistance = totalDistance / dataList.size();
        analysis.avgDuration = totalDuration / dataList.size();
        analysis.avgFarePerMile = totalFarePerMile / dataList.size();
        analysis.avgTipRate = totalTipRate / dataList.size();
        analysis.avgValueScore = totalValueScore / dataList.size();

        // 计算变异系数
        analysis.fareVariability = calculateVariability(dataList, "fare");
        analysis.tipVariability = calculateVariability(dataList, "tip");
        analysis.valueVariability = calculateVariability(dataList, "value");

        // 计算维度相关性
        analysis.dimensionCorrelation = calculateDimensionCorrelation(dataList);

        return analysis;
    }

    /**
     * 分析嵌套数据
     */
    private NestedAnalysis analyzeNestedData(List<NestedData> dataList) {
        NestedAnalysis analysis = new NestedAnalysis();

        Map<String, List<NestedData>> levelGroups = new HashMap<>();

        for (NestedData data : dataList) {
            levelGroups.computeIfAbsent(data.level, k -> new ArrayList<>()).add(data);
        }

        analysis.totalRecords = dataList.size();
        analysis.levelCount = levelGroups.size();

        // 分析各层级
        for (Map.Entry<String, List<NestedData>> entry : levelGroups.entrySet()) {
            String level = entry.getKey();
            List<NestedData> levelData = entry.getValue();

            double avgFare = levelData.stream().mapToDouble(d -> d.fareAmount).average().orElse(0);
            double avgEfficiency = levelData.stream().mapToDouble(d -> d.pricingEfficiency).average().orElse(0);
            double avgValue = levelData.stream().mapToDouble(d -> d.valueScore).average().orElse(0);

            if (level.contains("ROUTE")) {
                analysis.routeLevelAvgFare = avgFare;
                analysis.routeLevelAvgEfficiency = avgEfficiency;
                analysis.routeLevelCount = levelData.size();
            } else if (level.contains("TIME_CUSTOMER")) {
                analysis.timeCustomerLevelAvgFare = avgFare;
                analysis.timeCustomerLevelAvgValue = avgValue;
                analysis.timeCustomerLevelCount = levelData.size();
            } else if (level.contains("PRICING")) {
                analysis.pricingLevelAvgFare = avgFare;
                analysis.pricingLevelAvgEfficiency = avgEfficiency;
                analysis.pricingLevelCount = levelData.size();
            }
        }

        // 计算嵌套效果
        analysis.nestingEffectiveness = calculateNestingEffectiveness(levelGroups);

        return analysis;
    }

    /**
     * 分析条件数据
     */
    private ConditionalAnalysis analyzeConditionalData(List<ConditionalData> dataList) {
        ConditionalAnalysis analysis = new ConditionalAnalysis();

        Map<String, List<ConditionalData>> conditionGroups = new HashMap<>();

        for (ConditionalData data : dataList) {
            conditionGroups.computeIfAbsent(data.condition, k -> new ArrayList<>()).add(data);
        }

        analysis.totalRecords = dataList.size();
        analysis.conditionCount = conditionGroups.size();

        // 分析各条件
        for (Map.Entry<String, List<ConditionalData>> entry : conditionGroups.entrySet()) {
            String condition = entry.getKey();
            List<ConditionalData> condData = entry.getValue();

            double avgFare = condData.stream().mapToDouble(d -> d.fareAmount).average().orElse(0);
            double avgSurge = condData.stream().mapToDouble(d -> d.surgeMultiplier).average().orElse(0);
            double avgEfficiency = condData.stream().mapToDouble(d -> d.pricingEfficiency).average().orElse(0);

            if (condition.equals("PEAK_HIGH_VALUE")) {
                analysis.peakHighValueAvgFare = avgFare;
                analysis.peakHighValueAvgSurge = avgSurge;
                analysis.peakHighValueCount = condData.size();
            } else if (condition.equals("COMPETITIVE_LONG")) {
                analysis.competitiveLongAvgFare = avgFare;
                analysis.competitiveLongAvgEfficiency = avgEfficiency;
                analysis.competitiveLongCount = condData.size();
            } else if (condition.equals("PREMIUM_CONGESTION")) {
                analysis.premiumCongestionAvgFare = avgFare;
                analysis.premiumCongestionAvgSurge = avgSurge;
                analysis.premiumCongestionCount = condData.size();
            } else if (condition.equals("WEEKDAY_ECONOMY")) {
                analysis.weekdayEconomyAvgFare = avgFare;
                analysis.weekdayEconomyAvgEfficiency = avgEfficiency;
                analysis.weekdayEconomyCount = condData.size();
            }
        }

        // 计算条件效果
        analysis.conditionalEffectiveness = calculateConditionalEffectiveness(conditionGroups);

        return analysis;
    }

    /**
     * 分析滑动窗口数据
     */
    private SlidingWindowAnalysis analyzeSlidingWindowData(List<SlidingWindowData> dataList) {
        SlidingWindowAnalysis analysis = new SlidingWindowAnalysis();

        Map<String, List<SlidingWindowData>> windowGroups = new HashMap<>();

        for (SlidingWindowData data : dataList) {
            windowGroups.computeIfAbsent(data.windowType, k -> new ArrayList<>()).add(data);
        }

        analysis.totalRecords = dataList.size();
        analysis.windowTypeCount = windowGroups.size();

        // 分析各窗口类型
        for (Map.Entry<String, List<SlidingWindowData>> entry : windowGroups.entrySet()) {
            String windowType = entry.getKey();
            List<SlidingWindowData> windowData = entry.getValue();

            double avgFare = windowData.stream().mapToDouble(d -> d.fareAmount).average().orElse(0);
            double avgSurge = windowData.stream().mapToDouble(d -> d.surgeMultiplier).average().orElse(0);
            double avgValue = windowData.stream().mapToDouble(d -> d.valueScore).average().orElse(0);

            if (windowType.equals("HOUR")) {
                analysis.hourWindowAvgFare = avgFare;
                analysis.hourWindowAvgSurge = avgSurge;
                analysis.hourWindowCount = windowData.size();
            } else if (windowType.equals("PERIOD")) {
                analysis.periodWindowAvgFare = avgFare;
                analysis.periodWindowAvgValue = avgValue;
                analysis.periodWindowCount = windowData.size();
            } else if (windowType.equals("DAYTYPE")) {
                analysis.dayTypeWindowAvgFare = avgFare;
                analysis.dayTypeWindowAvgValue = avgValue;
                analysis.dayTypeWindowCount = windowData.size();
            }
        }

        // 计算窗口平滑度
        analysis.windowSmoothness = calculateWindowSmoothness(windowGroups);

        // 计算趋势强度
        analysis.trendStrength = calculateTrendStrength(dataList);

        return analysis;
    }

    /**
     * 分析分层数据
     */
    private HierarchicalAnalysis analyzeHierarchicalData(List<HierarchicalData> dataList) {
        HierarchicalAnalysis analysis = new HierarchicalAnalysis();

        Map<String, List<HierarchicalData>> levelGroups = new HashMap<>();

        for (HierarchicalData data : dataList) {
            levelGroups.computeIfAbsent(data.level, k -> new ArrayList<>()).add(data);
        }

        analysis.totalRecords = dataList.size();
        analysis.hierarchyLevels = levelGroups.size();

        // 分析各层级
        for (Map.Entry<String, List<HierarchicalData>> entry : levelGroups.entrySet()) {
            String level = entry.getKey();
            List<HierarchicalData> levelData = entry.getValue();

            double avgFare = levelData.stream().mapToDouble(d -> d.fareAmount).average().orElse(0);
            double avgTipRate = levelData.stream().mapToDouble(d -> d.tipRate).average().orElse(0);
            double avgValue = levelData.stream().mapToDouble(d -> d.valueScore).average().orElse(0);

            if (level.equals("GLOBAL")) {
                analysis.globalAvgFare = avgFare;
                analysis.globalAvgTipRate = avgTipRate;
                analysis.globalAvgValue = avgValue;
                analysis.globalCount = levelData.size();
            } else if (level.equals("TIME_TYPE")) {
                analysis.timeTypeAvgFare = avgFare;
                analysis.timeTypeAvgValue = avgValue;
                analysis.timeTypeCount = levelData.size();
            } else if (level.equals("DISTANCE_PRICING")) {
                analysis.distancePricingAvgFare = avgFare;
                analysis.distancePricingAvgValue = avgValue;
                analysis.distancePricingCount = levelData.size();
            } else if (level.equals("CUSTOMER")) {
                analysis.customerAvgFare = avgFare;
                analysis.customerAvgTipRate = avgTipRate;
                analysis.customerCount = levelData.size();
            } else if (level.equals("ROUTE")) {
                analysis.routeAvgFare = avgFare;
                analysis.routeAvgValue = avgValue;
                analysis.routeCount = levelData.size();
            }
        }

        // 计算层次一致性
        analysis.hierarchyConsistency = calculateHierarchyConsistency(levelGroups);

        // 计算聚合效率
        analysis.aggregationEfficiency = calculateAggregationEfficiency(levelGroups);

        return analysis;
    }

    /**
     * 计算变异系数
     */
    private double calculateVariability(List<MultiDimensionalData> dataList, String metric) {
        if (dataList.size() < 2) return 0;

        double[] values = new double[dataList.size()];
        for (int i = 0; i < dataList.size(); i++) {
            switch (metric) {
                case "fare":
                    values[i] = dataList.get(i).fareAmount;
                    break;
                case "tip":
                    values[i] = dataList.get(i).tipRate;
                    break;
                case "value":
                    values[i] = dataList.get(i).valueScore;
                    break;
                default:
                    values[i] = 0;
            }
        }

        double mean = Arrays.stream(values).average().orElse(0);
        double variance = Arrays.stream(values).map(x -> Math.pow(x - mean, 2)).average().orElse(0);
        double stdDev = Math.sqrt(variance);

        return mean != 0 ? stdDev / mean : 0;
    }

    /**
     * 计算维度相关性
     */
    private double calculateDimensionCorrelation(List<MultiDimensionalData> dataList) {
        if (dataList.size() < 2) return 0;

        // 计算费用与价值评分的相关性
        double[] fares = dataList.stream().mapToDouble(d -> d.fareAmount).toArray();
        double[] values = dataList.stream().mapToDouble(d -> d.valueScore).toArray();

        return calculateCorrelation(fares, values);
    }

    /**
     * 计算嵌套效果
     */
    private double calculateNestingEffectiveness(Map<String, List<NestedData>> levelGroups) {
        if (levelGroups.size() < 2) return 0;

        // 基于不同层级的数据分布计算嵌套效果
        double totalVariance = 0;
        int levelCount = 0;

        for (List<NestedData> levelData : levelGroups.values()) {
            if (levelData.size() > 1) {
                double[] fares = levelData.stream().mapToDouble(d -> d.fareAmount).toArray();
                double mean = Arrays.stream(fares).average().orElse(0);
                double variance = Arrays.stream(fares).map(x -> Math.pow(x - mean, 2)).average().orElse(0);
                totalVariance += variance;
                levelCount++;
            }
        }

        return levelCount > 0 ? 1.0 / (1.0 + totalVariance / levelCount) : 0;
    }

    /**
     * 计算条件效果
     */
    private double calculateConditionalEffectiveness(Map<String, List<ConditionalData>> conditionGroups) {
        if (conditionGroups.isEmpty()) return 0;

        // 基于条件筛选的精确度计算效果
        double totalPrecision = 0;

        for (Map.Entry<String, List<ConditionalData>> entry : conditionGroups.entrySet()) {
            List<ConditionalData> condData = entry.getValue();
            if (!condData.isEmpty()) {
                double avgEfficiency = condData.stream().mapToDouble(d -> d.pricingEfficiency).average().orElse(0);
                totalPrecision += Math.min(avgEfficiency, 2.0); // 限制在合理范围内
            }
        }

        return totalPrecision / conditionGroups.size();
    }

    /**
     * 计算窗口平滑度
     */
    private double calculateWindowSmoothness(Map<String, List<SlidingWindowData>> windowGroups) {
        double totalSmoothness = 0;
        int groupCount = 0;

        for (List<SlidingWindowData> windowData : windowGroups.values()) {
            if (windowData.size() > 2) {
                // 计算相邻数据点的变化率
                double totalChange = 0;
                for (int i = 1; i < windowData.size(); i++) {
                    double change = Math.abs(windowData.get(i).fareAmount - windowData.get(i-1).fareAmount);
                    totalChange += change;
                }
                double avgChange = totalChange / (windowData.size() - 1);
                double smoothness = 1.0 / (1.0 + avgChange);
                totalSmoothness += smoothness;
                groupCount++;
            }
        }

        return groupCount > 0 ? totalSmoothness / groupCount : 0;
    }

    /**
     * 计算趋势强度
     */
    private double calculateTrendStrength(List<SlidingWindowData> dataList) {
        if (dataList.size() < 3) return 0;

        // 按时间排序
        dataList.sort((a, b) -> Integer.compare(a.pickupHour, b.pickupHour));

        // 计算线性趋势
        double[] x = new double[dataList.size()];
        double[] y = new double[dataList.size()];

        for (int i = 0; i < dataList.size(); i++) {
            x[i] = i;
            y[i] = dataList.get(i).fareAmount;
        }

        return Math.abs(calculateCorrelation(x, y));
    }

    /**
     * 计算层次一致性
     */
    private double calculateHierarchyConsistency(Map<String, List<HierarchicalData>> levelGroups) {
        if (levelGroups.size() < 2) return 0;

        // 计算不同层级间的一致性
        double totalConsistency = 0;
        int comparisonCount = 0;

        List<String> levels = new ArrayList<>(levelGroups.keySet());
        for (int i = 0; i < levels.size(); i++) {
            for (int j = i + 1; j < levels.size(); j++) {
                List<HierarchicalData> level1 = levelGroups.get(levels.get(i));
                List<HierarchicalData> level2 = levelGroups.get(levels.get(j));

                if (!level1.isEmpty() && !level2.isEmpty()) {
                    double avg1 = level1.stream().mapToDouble(d -> d.fareAmount).average().orElse(0);
                    double avg2 = level2.stream().mapToDouble(d -> d.fareAmount).average().orElse(0);

                    double consistency = 1.0 / (1.0 + Math.abs(avg1 - avg2) / Math.max(avg1, avg2));
                    totalConsistency += consistency;
                    comparisonCount++;
                }
            }
        }

        return comparisonCount > 0 ? totalConsistency / comparisonCount : 0;
    }

    /**
     * 计算聚合效率
     */
    private double calculateAggregationEfficiency(Map<String, List<HierarchicalData>> levelGroups) {
        if (levelGroups.isEmpty()) return 0;

        // 基于数据压缩比和信息保留度计算聚合效率
        int totalRecords = levelGroups.values().stream().mapToInt(List::size).sum();
        int levelCount = levelGroups.size();

        // 计算压缩比
        double compressionRatio = (double) levelCount / totalRecords;

        // 计算信息保留度（基于方差保留）
        double totalVariance = 0;
        double levelVariance = 0;

        for (List<HierarchicalData> levelData : levelGroups.values()) {
            if (levelData.size() > 1) {
                double[] fares = levelData.stream().mapToDouble(d -> d.fareAmount).toArray();
                double mean = Arrays.stream(fares).average().orElse(0);
                double variance = Arrays.stream(fares).map(x -> Math.pow(x - mean, 2)).average().orElse(0);
                totalVariance += variance * levelData.size();
                levelVariance += variance;
            }
        }

        double informationRetention = totalRecords > 0 ? 1.0 - (levelVariance / totalVariance) : 0;

        // 综合效率 = 压缩比 * 信息保留度
        return compressionRatio * informationRetention;
    }

    /**
     * 计算相关系数
     */
    private double calculateCorrelation(double[] x, double[] y) {
        if (x.length != y.length || x.length < 2) return 0;

        double meanX = Arrays.stream(x).average().orElse(0);
        double meanY = Arrays.stream(y).average().orElse(0);

        double numerator = 0;
        double sumXSquare = 0;
        double sumYSquare = 0;

        for (int i = 0; i < x.length; i++) {
            double deltaX = x[i] - meanX;
            double deltaY = y[i] - meanY;
            numerator += deltaX * deltaY;
            sumXSquare += deltaX * deltaX;
            sumYSquare += deltaY * deltaY;
        }

        double denominator = Math.sqrt(sumXSquare * sumYSquare);
        return denominator != 0 ? numerator / denominator : 0;
    }

    /**
     * 输出多维度分析结果
     */
    private void outputMultiDimensionalAnalysis(String aggregateKey, MultiDimensionalAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("多维度聚合|记录数:%d|平均费用:%.2f|平均小费:%.2f|平均总额:%.2f|平均距离:%.2f|平均时长:%.2f|平均每英里费用:%.2f|平均小费率:%.3f|平均价值评分:%.2f|费用变异性:%.3f|小费变异性:%.3f|价值变异性:%.3f|维度相关性:%.3f",
                analysis.recordCount, analysis.avgFare, analysis.avgTip, analysis.avgTotalAmount,
                analysis.avgDistance, analysis.avgDuration, analysis.avgFarePerMile,
                analysis.avgTipRate, analysis.avgValueScore, analysis.fareVariability,
                analysis.tipVariability, analysis.valueVariability, analysis.dimensionCorrelation);

        context.write(new Text("多维度分析_" + aggregateKey), new Text(analysisInfo));
    }

    /**
     * 输出嵌套分析结果
     */
    private void outputNestedAnalysis(String nestedKey, NestedAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("嵌套聚合|总记录:%d|层级数:%d|路线层级数:%d|路线层级平均费用:%.2f|路线层级平均效率:%.3f|时间客户层级数:%d|时间客户层级平均费用:%.2f|时间客户层级平均价值:%.2f|定价层级数:%d|定价层级平均费用:%.2f|定价层级平均效率:%.3f|嵌套效果:%.3f",
                analysis.totalRecords, analysis.levelCount, analysis.routeLevelCount,
                analysis.routeLevelAvgFare, analysis.routeLevelAvgEfficiency,
                analysis.timeCustomerLevelCount, analysis.timeCustomerLevelAvgFare,
                analysis.timeCustomerLevelAvgValue, analysis.pricingLevelCount,
                analysis.pricingLevelAvgFare, analysis.pricingLevelAvgEfficiency,
                analysis.nestingEffectiveness);

        context.write(new Text("嵌套分析_" + nestedKey), new Text(analysisInfo));
    }

    /**
     * 输出条件分析结果
     */
    private void outputConditionalAnalysis(String conditionKey, ConditionalAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("条件聚合|总记录:%d|条件数:%d|高峰高价值数:%d|高峰高价值平均费用:%.2f|高峰高价值平均激增:%.2f|竞争长途数:%d|竞争长途平均费用:%.2f|竞争长途平均效率:%.3f|高端拥堵数:%d|高端拥堵平均费用:%.2f|高端拥堵平均激增:%.2f|工作日经济数:%d|工作日经济平均费用:%.2f|工作日经济平均效率:%.3f|条件效果:%.3f",
                analysis.totalRecords, analysis.conditionCount, analysis.peakHighValueCount,
                analysis.peakHighValueAvgFare, analysis.peakHighValueAvgSurge,
                analysis.competitiveLongCount, analysis.competitiveLongAvgFare,
                analysis.competitiveLongAvgEfficiency, analysis.premiumCongestionCount,
                analysis.premiumCongestionAvgFare, analysis.premiumCongestionAvgSurge,
                analysis.weekdayEconomyCount, analysis.weekdayEconomyAvgFare,
                analysis.weekdayEconomyAvgEfficiency, analysis.conditionalEffectiveness);

        context.write(new Text("条件分析_" + conditionKey), new Text(analysisInfo));
    }

    /**
     * 输出滑动窗口分析结果
     */
    private void outputSlidingWindowAnalysis(String windowKey, SlidingWindowAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("滑动窗口|总记录:%d|窗口类型数:%d|小时窗口数:%d|小时窗口平均费用:%.2f|小时窗口平均激增:%.2f|时段窗口数:%d|时段窗口平均费用:%.2f|时段窗口平均价值:%.2f|日类型窗口数:%d|日类型窗口平均费用:%.2f|日类型窗口平均价值:%.2f|窗口平滑度:%.3f|趋势强度:%.3f",
                analysis.totalRecords, analysis.windowTypeCount, analysis.hourWindowCount,
                analysis.hourWindowAvgFare, analysis.hourWindowAvgSurge,
                analysis.periodWindowCount, analysis.periodWindowAvgFare,
                analysis.periodWindowAvgValue, analysis.dayTypeWindowCount,
                analysis.dayTypeWindowAvgFare, analysis.dayTypeWindowAvgValue,
                analysis.windowSmoothness, analysis.trendStrength);

        context.write(new Text("滑动窗口分析_" + windowKey), new Text(analysisInfo));
    }

    /**
     * 输出分层分析结果
     */
    private void outputHierarchicalAnalysis(String hierarchyKey, HierarchicalAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("分层聚合|总记录:%d|层级数:%d|全局数:%d|全局平均费用:%.2f|全局平均小费率:%.3f|全局平均价值:%.2f|时间类型数:%d|时间类型平均费用:%.2f|时间类型平均价值:%.2f|距离定价数:%d|距离定价平均费用:%.2f|距离定价平均价值:%.2f|客户数:%d|客户平均费用:%.2f|客户平均小费率:%.3f|路线数:%d|路线平均费用:%.2f|路线平均价值:%.2f|层次一致性:%.3f|聚合效率:%.3f",
                analysis.totalRecords, analysis.hierarchyLevels, analysis.globalCount,
                analysis.globalAvgFare, analysis.globalAvgTipRate, analysis.globalAvgValue,
                analysis.timeTypeCount, analysis.timeTypeAvgFare, analysis.timeTypeAvgValue,
                analysis.distancePricingCount, analysis.distancePricingAvgFare,
                analysis.distancePricingAvgValue, analysis.customerCount,
                analysis.customerAvgFare, analysis.customerAvgTipRate, analysis.routeCount,
                analysis.routeAvgFare, analysis.routeAvgValue, analysis.hierarchyConsistency,
                analysis.aggregationEfficiency);

        context.write(new Text("分层分析_" + hierarchyKey), new Text(analysisInfo));
    }

    // 解析方法
    private MultiDimensionalData parseMultiDimensionalData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 9) {
                MultiDimensionalData mdData = new MultiDimensionalData();
                mdData.dimension = fields[0];
                mdData.fareAmount = Double.parseDouble(fields[1]);
                mdData.tipAmount = Double.parseDouble(fields[2]);
                mdData.totalAmount = Double.parseDouble(fields[3]);
                mdData.tripDistance = Double.parseDouble(fields[4]);
                mdData.tripDuration = Double.parseDouble(fields[5]);
                mdData.farePerMile = Double.parseDouble(fields[6]);
                mdData.tipRate = Double.parseDouble(fields[7]);
                mdData.valueScore = Double.parseDouble(fields[8]);
                return mdData;
            }
        } catch (Exception e) {
            System.err.println("解析多维度数据错误: " + data);
        }
        return null;
    }

    private NestedData parseNestedData(String level, String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 13) {
                NestedData nestedData = new NestedData();
                nestedData.level = fields[0];
                nestedData.dim1 = fields[1];
                nestedData.dim2 = fields[2];
                nestedData.dim3 = fields[3];
                nestedData.dim4 = fields[4];
                nestedData.fareAmount = Double.parseDouble(fields[5]);
                nestedData.tipAmount = Double.parseDouble(fields[6]);
                nestedData.totalAmount = Double.parseDouble(fields[7]);
                nestedData.tripDistance = Double.parseDouble(fields[8]);
                nestedData.farePerMile = Double.parseDouble(fields[9]);
                nestedData.surgeMultiplier = Double.parseDouble(fields[10]);
                nestedData.pricingEfficiency = Double.parseDouble(fields[11]);
                nestedData.valueScore = Double.parseDouble(fields[12]);
                return nestedData;
            }
        } catch (Exception e) {
            System.err.println("解析嵌套数据错误: " + data);
        }
        return null;
    }

    private ConditionalData parseConditionalData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 14) {
                ConditionalData condData = new ConditionalData();
                condData.condition = fields[0];
                condData.pickupHour = Integer.parseInt(fields[1]);
                condData.timePeriod = fields[2];
                condData.dayType = fields[3];
                condData.distanceCategory = fields[4];
                condData.pricingTier = fields[5];
                condData.fareAmount = Double.parseDouble(fields[6]);
                condData.tipAmount = Double.parseDouble(fields[7]);
                condData.totalAmount = Double.parseDouble(fields[8]);
                condData.tripDistance = Double.parseDouble(fields[9]);
                condData.farePerMile = Double.parseDouble(fields[10]);
                condData.tipRate = Double.parseDouble(fields[11]);
                condData.surgeMultiplier = Double.parseDouble(fields[12]);
                condData.pricingEfficiency = Double.parseDouble(fields[13]);
                return condData;
            }
        } catch (Exception e) {
            System.err.println("解析条件数据错误: " + data);
        }
        return null;
    }

    private SlidingWindowData parseSlidingWindowData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 12) {
                SlidingWindowData swData = new SlidingWindowData();
                swData.windowType = fields[0];
                swData.pickupDate = fields[1];
                swData.pickupHour = Integer.parseInt(fields[2]);
                swData.timePeriod = fields[3];
                swData.dayType = fields[4];
                swData.fareAmount = Double.parseDouble(fields[5]);
                swData.tipAmount = Double.parseDouble(fields[6]);
                swData.totalAmount = Double.parseDouble(fields[7]);
                swData.tripDistance = Double.parseDouble(fields[8]);
                swData.farePerMile = Double.parseDouble(fields[9]);
                swData.surgeMultiplier = Double.parseDouble(fields[10]);
                swData.valueScore = Double.parseDouble(fields[11]);
                return swData;
            }
        } catch (Exception e) {
            System.err.println("解析滑动窗口数据错误: " + data);
        }
        return null;
    }

    private HierarchicalData parseHierarchicalData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 12) {
                HierarchicalData hData = new HierarchicalData();
                hData.level = fields[0];
                hData.dim1 = fields[1];
                hData.dim2 = fields[2];
                hData.dim3 = fields[3];
                hData.dim4 = fields[4];
                hData.fareAmount = Double.parseDouble(fields[5]);
                hData.tipAmount = Double.parseDouble(fields[6]);
                hData.totalAmount = Double.parseDouble(fields[7]);
                hData.tripDistance = Double.parseDouble(fields[8]);
                hData.farePerMile = Double.parseDouble(fields[9]);
                hData.tipRate = Double.parseDouble(fields[10]);
                hData.valueScore = Double.parseDouble(fields[11]);
                return hData;
            }
        } catch (Exception e) {
            System.err.println("解析分层数据错误: " + data);
        }
        return null;
    }

    // 内部类定义
    private static class MultiDimensionalData {
        String dimension;
        double fareAmount, tipAmount, totalAmount, tripDistance, tripDuration;
        double farePerMile, tipRate, valueScore;
    }

    private static class NestedData {
        String level, dim1, dim2, dim3, dim4;
        double fareAmount, tipAmount, totalAmount, tripDistance, farePerMile;
        double surgeMultiplier, pricingEfficiency, valueScore;
    }

    private static class ConditionalData {
        String condition, timePeriod, dayType, distanceCategory, pricingTier;
        int pickupHour;
        double fareAmount, tipAmount, totalAmount, tripDistance, farePerMile;
        double tipRate, surgeMultiplier, pricingEfficiency;
    }

    private static class SlidingWindowData {
        String windowType, pickupDate, timePeriod, dayType;
        int pickupHour;
        double fareAmount, tipAmount, totalAmount, tripDistance, farePerMile;
        double surgeMultiplier, valueScore;
    }

    private static class HierarchicalData {
        String level, dim1, dim2, dim3, dim4;
        double fareAmount, tipAmount, totalAmount, tripDistance, farePerMile;
        double tipRate, valueScore;
    }

    private static class MultiDimensionalAnalysis {
        int recordCount;
        double avgFare, avgTip, avgTotalAmount, avgDistance, avgDuration;
        double avgFarePerMile, avgTipRate, avgValueScore;
        double fareVariability, tipVariability, valueVariability, dimensionCorrelation;
    }

    private static class NestedAnalysis {
        int totalRecords, levelCount;
        int routeLevelCount, timeCustomerLevelCount, pricingLevelCount;
        double routeLevelAvgFare, routeLevelAvgEfficiency;
        double timeCustomerLevelAvgFare, timeCustomerLevelAvgValue;
        double pricingLevelAvgFare, pricingLevelAvgEfficiency;
        double nestingEffectiveness;
    }

    private static class ConditionalAnalysis {
        int totalRecords, conditionCount;
        int peakHighValueCount, competitiveLongCount, premiumCongestionCount, weekdayEconomyCount;
        double peakHighValueAvgFare, peakHighValueAvgSurge;
        double competitiveLongAvgFare, competitiveLongAvgEfficiency;
        double premiumCongestionAvgFare, premiumCongestionAvgSurge;
        double weekdayEconomyAvgFare, weekdayEconomyAvgEfficiency;
        double conditionalEffectiveness;
    }

    private static class SlidingWindowAnalysis {
        int totalRecords, windowTypeCount;
        int hourWindowCount, periodWindowCount, dayTypeWindowCount;
        double hourWindowAvgFare, hourWindowAvgSurge;
        double periodWindowAvgFare, periodWindowAvgValue;
        double dayTypeWindowAvgFare, dayTypeWindowAvgValue;
        double windowSmoothness, trendStrength;
    }

    private static class HierarchicalAnalysis {
        int totalRecords, hierarchyLevels;
        int globalCount, timeTypeCount, distancePricingCount, customerCount, routeCount;
        double globalAvgFare, globalAvgTipRate, globalAvgValue;
        double timeTypeAvgFare, timeTypeAvgValue;
        double distancePricingAvgFare, distancePricingAvgValue;
        double customerAvgFare, customerAvgTipRate;
        double routeAvgFare, routeAvgValue;
        double hierarchyConsistency, aggregationEfficiency;
    }
}
